package com.zhike.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.zhike.utils.DimUtils;
import com.zhike.utils.DruidDSUtils;
import com.zhike.utils.ThreadPoolUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {
    public DimAsyncFunction() {
    }

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    private String tableName;
    private String key;
    private DruidDataSource dataSource = null;
    private ThreadPoolExecutor threadPoolExecutor = null;


    @Override
    public void open(Configuration parameters) throws Exception {
        //获取连接池和线程池
        dataSource = DruidDSUtils.createDataSource();
        threadPoolExecutor = ThreadPoolUtils.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        //来一条数据创建一个线程
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    //获取连接
                    DruidPooledConnection connection = dataSource.getConnection();
                    //查询维表数据
                    String key = getKey(input);
                    JSONObject dimInfo = DimUtils.getDimInfo(connection, tableName, key);

                    //将维度信息补充到当前数据
                    if (dimInfo != null) {
                        join(input, dimInfo);
                    }

                    //关闭连接
                    connection.close();
                    //返回结果
                    resultFuture.complete(Collections.singletonList(input));
                } catch (Exception e) {
                    System.out.println("关联维表失败");
                }
            }


        });

    }


    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {

    }
}